// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: daniel

#include "coin2/base/request/http2/connection.h"

#include <utility>

#include <boost/asio/buffer.hpp>

namespace coin2::base::request::http2 {

Connection::Connection(
    boost::asio::io_context* io_ctx,
    boost::asio::ssl::context* tls_ctx,
    const std::string& host,
    int port,
    const std::string& bind_ip)
    : io_ctx_{io_ctx},
      tls_ctx_{tls_ctx},
      host_{host},
      port_{port},
      bind_ip_{bind_ip},
      deadline_timer_{*io_ctx},
      session_{nullptr} {
  conn_start_ts_ = GetCurrentTimestamp();
  conn_id_debug_ = conn_start_ts_;
  LOG(INFO) << "[http2] new session is created [" << conn_id_debug_ << "]. " << this;
  InitSsl();
}

Connection::~Connection() {
  Shutdown();
  LOG(INFO) << "[http2] session is terminated [" << conn_id_debug_
            << "]. total recv: " << RequestSent() << ". " << this;
}

void Connection::InitSsl() {
  if (!tls_ctx_) {
    return;
  }
  boost::system::error_code ec;
  tls_ctx_->set_default_verify_paths();
  nghttp2::asio_http2::client::configure_tls_context(ec, *tls_ctx_);
  if (ec) {
    LOG(ERROR) << "[http2] error on onfigure tls context, " << ec.message() << ". " << this;
  }
}

void Connection::Connect() { DoConnect(); }

void Connection::Suspend() {
  Shutdown();
  if (session_) {
    session_.reset();
  }
}

void Connection::DoConnect() {
  is_connected_ = false;
  IsConnecting(true);
  LOG(INFO) << "[http2] connecting to service: " << host_ << ":" << port_ << ". " << this;
  switch (port_) {
    case 80:
      ConnectWithNonSsl();
      break;
    case 443:
      ConnectWithSsl();
      break;
    default:
      CHECK(false) << "unsupported port" << port_;
      break;
  }
  if (!session_) {
    LOG(ERROR) << "[http2] failed to connect due to http2 session is unavailable. " << this;
    IsConnecting(false);
    return;
  }
  session_->on_connect(
      std::bind(&Connection::OnConnect, shared_from_this(), std::placeholders::_1));
  session_->on_error(std::bind(&Connection::OnError, shared_from_this(), std::placeholders::_1));
}

void Connection::Shutdown() {
  bool is_connected = is_connected_ ? true : false;
  is_connected_ = false;
  if (session_ && is_connected) {
    session_->shutdown();
  }
}

void Connection::ConnectWithNonSsl() {
  use_ssl_ = false;
  if (!bind_ip_.empty()) {
    boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(bind_ip_), 0);
    session_.reset(new NgHttp2Session(*io_ctx_, endpoint, host_, "http"));
  } else {
    session_.reset(new NgHttp2Session(*io_ctx_, host_, "http"));
  }
}

void Connection::ConnectWithSsl() {
  use_ssl_ = true;
  if (!bind_ip_.empty()) {
    boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(bind_ip_), 0);
    session_.reset(new NgHttp2Session(*io_ctx_, *tls_ctx_, endpoint, host_, "https"));
  } else {
    session_.reset(new NgHttp2Session(*io_ctx_, *tls_ctx_, host_, "https"));
  }
}

// which will be invoked after connection is established.
void Connection::OnConnect(boost::asio::ip::tcp::resolver::iterator endpoint_it) {
  is_connected_ = true;
  IsConnecting(false);
  hit_fail_tolerance_ = false;
  LOG(INFO) << "[http2] connected to service: " << host_ << ":" << port_ << ". " << this;
}

// which will be invoked there is connection level error and session is terminated.
void Connection::OnError(const boost::system::error_code& ec) {
  is_connected_ = false;
  IsConnecting(false);
  LOG(ERROR) << "[http2] disconnected from service: " << host_ << ":" << port_
             << ", reason=" << (ec ? ec.message() : "unknown error") << ". " << this;
  Reconnect();
}

void Connection::Reconnect() {
  if (IsConnecting()) {
    return;
  }
  Shutdown();
  deadline_timer_.expires_from_now(boost::posix_time::seconds(reconnect_interval_));
  deadline_timer_.async_wait([this, self = shared_from_this()](const boost::system::error_code& e) {
    LOG(INFO) << "[http2] reconnect service: " << host_ << ":" << port_ << ". " << this;
    DoConnect();
  });
}

void Connection::Submit(std::weak_ptr<Http2AsioContext> context, CallbackType callback) {
  auto ctx = context.lock();
  if (!ctx) {
    LOG(ERROR) << "[http2] failed to send request, context is expired";
    return;
  }
  ++request_sent_;
  DLOG(INFO) << "[http2] total of requests submitted: " << request_sent_
             << ", session_id=" << conn_id_debug_;
  if (!IsConnected()) {
    OnClose(ctx, std::move(callback), Nghttp2ErrorCode::NGHTTP2_CONNECT_ERROR);
    return;
  }
  using Http2RequestParser = coin2::base::request::http2::RequestParser;
  Http2RequestParser hrp{use_ssl_, ctx->inner_req, ctx->weight};
  boost::system::error_code ec;
  if (!session_) {
    LOG(ERROR) << "[http2] failed to submit due to http2 session is unavailable. " << this;
    ec = boost::system::errc::make_error_code(boost::system::errc::too_many_files_open);
    OnClose(ctx, std::move(callback), ec.value());
    return;
  }
  auto req =
      session_->submit(ec, hrp.Method(), hrp.Uri(), hrp.Payload(), hrp.Headers(), hrp.Priority());
  if (ec) {
    LOG(ERROR) << "[http2] failed to submit, reason=" << ec.message() << "[" << ec.value() << "]. "
               << this;
    OnClose(ctx, std::move(callback), ec.value());
    return;
  }
  ctx->outer_req = req;
  req->on_response(
      std::bind(&Connection::OnResponse, shared_from_this(), context, std::placeholders::_1));
  req->on_close(std::bind(
      &Connection::OnClose,
      shared_from_this(),
      context,
      std::move(callback),
      std::placeholders::_1));
}

void Connection::OnResponse(std::weak_ptr<Http2AsioContext> context, const NgHttp2Response& res) {
  auto ctx = context.lock();
  if (!ctx) {
    LOG(ERROR) << "[http2] failed to process response head, context is expired";
    return;
  }
  auto& response = ctx->inner_res;
  response.result(res.status_code());
  auto& header = response.base();
  for (auto& kv : res.header()) {
    header.set(kv.first, kv.second.value);
  }
  DLOG(INFO) << "[http2] received response header [" << ctx->request_id
             << "], status_code=" << ctx->inner_res.result();
  res.on_data(std::bind(
      &Connection::OnData,
      shared_from_this(),
      context,
      std::placeholders::_1,
      std::placeholders::_2));
}

void Connection::OnData(
    std::weak_ptr<Http2AsioContext> context,
    const uint8_t* data,
    std::size_t len) {
  auto ctx = context.lock();
  if (!ctx) {
    LOG(ERROR) << "[http2] failed to process response data, context expired";
    return;
  }
  size_t n = boost::asio::buffer_copy(ctx->buff.prepare(len), boost::asio::buffer(data, len));
  ctx->buff.commit(n);
}

void Connection::OnClose(
    std::weak_ptr<Http2AsioContext> context,
    CallbackType callback,
    uint32_t error_code) {
  ++request_recv_;
  DLOG(INFO) << "[http2] request closed with error_code="
             << Nghttp2ErrorFormater::GetErrorString(error_code);
  DLOG(INFO) << "[http2] total of requests closed: " << request_recv_
             << ", session_id=" << conn_id_debug_;
  if (auto ctx = context.lock()) {
    ctx->outer_req = nullptr;
    ctx->inner_res.body() = boost::beast::buffers_to_string(ctx->buff.data());
    ctx->inner_res.prepare_payload();
    callback(ctx, Nghttp2ErrorFormater::GetBoostError(error_code));
  }
  OnFail(error_code, 0 /* NO_ERROR */);
}

}  // namespace coin2::base::request::http2
